package com.baosight.xdata;

import com.alibaba.fastjson.JSON;
import com.baosight.xdata.dto.TssEventMessageRecord;
import com.baosight.xdata.redis.LettuceUtil;
import io.lettuce.core.StreamMessage;
import io.lettuce.core.XReadArgs;

import java.net.InetAddress;
import java.time.Duration;
import java.util.List;
import java.util.Map;

public class StreamConsumer implements Runnable{

    public void sync() throws Exception {
        List<StreamMessage<String, String>> list1;
        List<StreamMessage<String, String>> list2;
        String host=InetAddress.getLocalHost().getHostAddress();

        while (true){
            XReadArgs xReadArgs = new XReadArgs();
            xReadArgs.block(Duration.ofSeconds(10));
            xReadArgs.count(5);

            list1 = LettuceUtil.xreadGroupWithArgs("tss-stream:1","tss-group1",host,xReadArgs);
            /*list1.stream().forEach(a->{
                Map map = a.getBody();
                TssEventMessageRecord tssEventMessageRecord = JSON.parseObject(a.getBody().toString(), TssEventMessageRecord.class);
            });*/
            for (int i=0;i<list1.size();i++){
                StreamMessage streamMessage = list1.get(i);
                Map map = streamMessage.getBody();
                System.out.println(map.toString());
            }
            list2 = LettuceUtil.xreadGroupWithArgs("tss-stream:2","tss-group2",host,xReadArgs);

            if (list1.isEmpty() && list2.isEmpty()) {
                System.out.println("==============================");
                try {
                    Thread.sleep(200);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                continue;
            }
            LettuceUtil.xdel("tss-stream:1",list1);
            LettuceUtil.xdel("tss-stream:2",list2);

            LettuceUtil.xack("tss-stream:1",list1,"tss-group1");
            LettuceUtil.xack("tss-stream:2",list2,"tss-group2");
        }
    }
    @Override
    public void run() {
        try {
            sync();
        } catch (Exception e) {
            e.printStackTrace();
            System.exit(-1);
        }
    }
}
